package com.delayqueue.core.permanent;

import com.delayqueue.common.task.Task;
import com.delayqueue.core.permanent.executor.PermanentCompletedExecutor;
import com.delayqueue.core.permanent.executor.PermanentTaskExecutor;
import com.delayqueue.core.queue.QueueData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.File;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Component
public class PermanentHandler {

    @Autowired
    private PermanentConfig permanentConfig;
    @Autowired
    private PermanentFile permanentFile;
    @Autowired
    private PermanentConsumer permanentConsumer;
    @Autowired
    private PermanentWriter permanentWriter;

    public void init() {
        recoveryTaskQueue();
        if (permanentConfig.getEnabled()) {
            permanentWriter.init();
            if (permanentConfig.getAsync() || permanentConfig.getArchive()) {
                new Thread(permanentConsumer).start();
            }
            new Timer().scheduleAtFixedRate(new TimerTask() {
                @Override
                public void run() {
                    cleanTask();
                }
            }, permanentConfig.getCleanTaskInterval() * 1000 * 60, permanentConfig.getCleanTaskInterval() * 1000 * 60);
        } else {
            File dir = new File(permanentConfig.getDataDir());
            File[] fileList = dir.listFiles();
            if (fileList.length > 0) {
                for (File file : fileList) {
                    if (file.isFile()) {
                        file.delete();
                    }
                }
            }
        }
        permanentFile.deleteTempFile();
    }


    public void addTask(Task task) {
        if (permanentConfig.getEnabled()) {
            permanentConsumer.add(task, PermanentTaskExecutor.class);
        }
    }

    public void addCompleted(Task task) {
        if (permanentConfig.getEnabled()) {
            permanentConsumer.add(task, PermanentCompletedExecutor.class);
        }
    }

    /**
     * 将队列从文件恢复到内存
     */
    private void recoveryTaskQueue() {
        log.info("开始恢复数据");
        HashSet<Task> taskList = permanentFile.loadTaskList();
        if (!taskList.isEmpty()) {
            HashSet<String> completedCodeList = permanentFile.loadCompletedCodeList();
            List<Task> recoverList = taskList.stream().filter(o -> !completedCodeList.contains(o.getCode())).collect(Collectors.toList());
            log.info("共需要恢复{}条记录", recoverList.size());
            if (!recoverList.isEmpty()) {
                Iterator<Task> iterator = recoverList.iterator();
                while (iterator.hasNext()) {
                    QueueData.addTask(iterator.next());
                }
                log.info("数据恢复完毕");
            }
        }
    }

    private void cleanTask() {
        log.info("开始清理");
        permanentWriter.changeStatusToCleaning();
        HashSet<Task> taskList = permanentFile.loadTaskList();
        if (!taskList.isEmpty()) {
            HashSet<String> completedCodeList = permanentFile.loadCompletedCodeList();
            List<Task> resultTaskList = taskList.stream().filter(o -> !completedCodeList.contains(o.getCode())).collect(Collectors.toList());
            for (Task task : resultTaskList) {
                permanentWriter.writeTask(task);
            }
            permanentWriter.cancelStatusCleaning();
            log.info("清理完毕，共有{}个task，保留{}个task", taskList.size(), resultTaskList.size());
        } else {
            permanentFile.clearCompletedDataFile();
            permanentWriter.cancelStatusCleaning();
            log.info("无需清理");
        }
    }
}
